package NBA

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.Partitioner

object ChamptionCountRDD {
  def main(args: Array[String]): Unit = {
    getChamptionPartioner()
    getChamptionCount()
  }
  
  // 划分分区
  def getChamptionPartioner():Unit = {
    // 实例化
    val conf = new SparkConf().setAppName("getChamptionPartioner").setMaster("local[3]")
    val sc = new SparkContext(conf)
    
    // 读取清理后的数据
    val rdd = sc.textFile("A:/output3/nba_clear/part-00000")
                .map(_.split(","))
                .map(arr => {
                  var flag = ""
                  if(arr.length <= 5) flag = "N"
                  else flag = arr(5)
                  (flag,arr(2))
                })
                .partitionBy(new Mypartitioner(3))
                .map(arr => arr._2 + "," + 1)
                .saveAsTextFile("A:/output3/nba_comption_partition") 
    sc.stop()
  }
  
  // 统计不同分区的冠军获得次数
  def getChamptionCount():Unit = {
    // 实例化
    val conf = new SparkConf().setAppName("getChamptionCount").setMaster("local[1]")
    val sc = new SparkContext(conf)
    sc.textFile("A:/output3/nba_comption_partition/part-00000")
      .map(_.split(","))
      .map(arr => (arr(0), arr(1).toInt))
      .reduceByKey(_+_)
      .sortBy(_._2,false,1)
      .saveAsTextFile("A:/output3/nba_comption_partition_N")
      
    sc.textFile("A:/output3/nba_comption_partition/part-00001")
      .map(_.split(","))
      .map(arr => (arr(0), arr(1).toInt))
      .reduceByKey(_+_)
      .sortBy(_._2,false,1)
      .saveAsTextFile("A:/output3/nba_comption_partition_W")
      
    sc.textFile("A:/output3/nba_comption_partition/part-00002")
      .map(_.split(","))
      .map(arr => (arr(0), arr(1).toInt))
      .reduceByKey(_+_)
      .sortBy(_._2,false,1)
      .saveAsTextFile("A:/output3/nba_comption_partition_E")      
  }
  
  class Mypartitioner(num:Int) extends Partitioner{
    override def numPartitions:Int = num
    override def getPartition(key:Any):Int = {
      if(key == "N") 0
      else if(key == "W") 1
      else 2
    }
  }
  
}